package com.scientific.research.gobang.service.rocketmq;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.scientific.research.gobang.Utils.RedisUtil;
import com.scientific.research.gobang.mapper.UserMapper;
import com.scientific.research.gobang.pojo.RunCodeTable;
import com.scientific.research.gobang.pojo.User;
import com.scientific.research.gobang.service.CodeAndCodeService;
import com.scientific.research.gobang.service.RocketMqService;
import com.scientific.research.gobang.utils.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
@Slf4j
public class RocketMqServiceImpl implements RocketMqService {
    @Resource
    private UserMapper userMapper;

    @Resource
    private RedisUtil redisUtil;

    @Resource(name = "codeAndCodeServiceImpl")
    private CodeAndCodeService codeAndCodeService;

    //从配置文件中，获取客户端连接地址
    @Value("${rocketmq.url}")
    private String url;

    @Override
    public void dealMessage() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumergroup");
        consumer.setNamesrvAddr(url);

        try {
            consumer.subscribe("runCodeMessage", "*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, consumeConcurrentlyContext) -> {
            //获取用户1和用户2的个人信息
            String message = new String(msgs.get(0).getBody());
            RunCodeTable runMessage = JsonUtils.toBean(message, RunCodeTable.class);
            QueryWrapper<User> wrapper = new QueryWrapper<>();
            wrapper.eq("email", runMessage.getFightUser());
            User userTwo = userMapper.selectOne(wrapper);

            //开始执行代码
            codeAndCodeService.CodeAndCodeOnce(runMessage.getEmail(), runMessage.getLanguage(), runMessage.getCode(), userTwo.getEmail(), userTwo.getCodeLanguage(), userTwo.getCode());
            redisUtil.set(runMessage.getEmail() + "_runCode", (int) redisUtil.get(runMessage.getEmail() + "_runCode") - 1);
            //清空SubmitList的Redis缓存
            if (redisUtil.get(runMessage.getEmail() + "_submitList") != null) {
                redisUtil.del(runMessage.getEmail() + "_submitList");
            }
            log.info("代码运行完毕！");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        try {
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        log.info("Consumer Started.");
    }
}
